package com.bingxu.flink.state;

import com.bingxu.flink.bean.InputNumber;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @author :BingXu
 * @description :TODO 连续输入数字大于100
 * @date :2021/8/19 14:58
 * @modifier :
 */

public class KeyMapState {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 10086);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);

        DataStreamSource<String> sourceStream = env.socketTextStream("localhost", 9999);

        sourceStream
                .map(new MapFunction<String, InputNumber>() {
                    @Override
                    public InputNumber map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new InputNumber(fields[0], Integer.parseInt(fields[1]));

                    }
                })
                .keyBy(ele -> ele.getId())
                .process(new KeyedProcessFunction<String, InputNumber, String>() {

                    private MapState<Integer, String> mapState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<Integer, String>("mapState",
                                Integer.class, String.class));

                    }

                    @Override
                    public void processElement(InputNumber value, Context ctx, Collector<String> out) throws Exception {
                        if (!mapState.contains(value.getNum())) {
                            mapState.put(value.getNum(), value.toString());
                            out.collect(value.toString());
                        }

                    }
                }).print();
                

                    

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
